/*
 *
 * Copyright (c) 2020 The Raptor Authors. All rights reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

#include "src/windows/connection.h"
#include <string.h>

#include "raptor-lite/impl/endpoint.h"
#include "raptor-lite/impl/handler.h"
#include "raptor-lite/utils/log.h"
#include "raptor-lite/utils/useful.h"

#include "src/common/connection_id.h"
#include "src/common/endpoint_impl.h"
#include "src/common/socket_util.h"
#include "src/common/common.h"

#include "src/windows/socket_settings.h"

namespace raptor {

Connection::Connection(std::shared_ptr<EndpointImpl> obj)
    : service_(nullptr)
    , event_service_(nullptr)
    , proto_(nullptr)
    , rs_thread_(nullptr)
    , send_pending_(false)
    , index_(0) {

    memset(&send_overlapped_, 0, sizeof(send_overlapped_));
    memset(&recv_overlapped_, 0, sizeof(recv_overlapped_));

    endpoint_ = obj;

    next_package_size_ = 0;
}

Connection::~Connection() {}

void Connection::Init(int32_t index) {
    index_ = index;

    send_pending_ = false;

    send_overlapped_.event_type = internal::kSendEvent;
    recv_overlapped_.event_type = internal::kRecvEvent;

    // Save connection id
    send_overlapped_.user_id = endpoint_->connection_id_;
    recv_overlapped_.user_id = endpoint_->connection_id_;
}

void Connection::SetPollingThread(PollingThread *rs_thread) { rs_thread_ = rs_thread; }

bool Connection::SetTransferService(internal::NotificationTransferService *service,
                                    internal::EventReceivingService *event_service) {
    service_ = service;
    event_service_ = event_service;
    rs_thread_->Add(static_cast<SOCKET>(endpoint_->fd_), event_service_);
    return AsyncRecv();
}

void Connection::SetProtocol(ProtocolHandler *p) { proto_ = p; }
void Connection::DoHeartBeat() { service_->OnHeartBeat(endpoint_); }

void Connection::Shutdown(bool notify, raptor_error desc) {
    if (!endpoint_->IsValid()) {
        return;
    }

    rs_thread_->Delete(static_cast<SOCKET>(endpoint_->fd_));
    raptor_set_socket_shutdown(static_cast<SOCKET>(endpoint_->fd_));
    endpoint_->fd_ = static_cast<uint64_t>(INVALID_SOCKET);
    send_pending_ = false;

    if (notify) {
        service_->OnClosed(endpoint_, desc);
    }

    //_rcv_mtx.Lock();
    rcv_buffer_.ClearBuffer();
    //_rcv_mtx.Unlock();

    //_snd_mtx.Lock();
    snd_buffer_.ClearBuffer();
    //_snd_mtx.Unlock();
}

bool Connection::SendMsg(const Slice &s) {
    if (!endpoint_->IsValid()) return false;

    // AutoMutex g(&_snd_mtx);
    snd_buffer_.AddSlice(s);
    return AsyncSend();
}

constexpr size_t MAX_PACKAGE_SIZE = 131072;
constexpr size_t MAX_WSABUF_COUNT = 24;

bool Connection::AsyncSend() {
    if (snd_buffer_.Empty() || send_pending_) {
        return true;
    }

    WSABUF buffers[MAX_WSABUF_COUNT];
    size_t prepare_send_size = 0;

    send_pending_ = true;

    size_t i, count = snd_buffer_.SliceCount();
    for (i = 0; i < count && i < MAX_WSABUF_COUNT; i++) {
        const Slice &s = snd_buffer_[i];

        if (i != 0 && prepare_send_size + s.length() > MAX_PACKAGE_SIZE) {
            break;
        }

        buffers[i].buf = s.buffer();
        buffers[i].len = static_cast<ULONG>(s.length());
        prepare_send_size += s.length();
    }

    // https://docs.microsoft.com/en-us/windows/win32/api/winsock2/nf-winsock2-wsasend
    int ret = WSASend((SOCKET)endpoint_->fd_, buffers, (DWORD)i, NULL, 0,
                      &send_overlapped_.overlapped, NULL);

    if (ret == SOCKET_ERROR) {
        int error = WSAGetLastError();
        if (error != WSA_IO_PENDING) {
            send_pending_ = false;
            return false;
        }
    }
    return true;
}

bool Connection::AsyncRecv() {
    DWORD dwFlags = 0;
    WSABUF buffer;

    buffer.buf = NULL;
    buffer.len = 0;

    // https://docs.microsoft.com/en-us/windows/win32/api/winsock2/nf-winsock2-wsarecv
    int ret = WSARecv((SOCKET)endpoint_->fd_, &buffer, 1, NULL, &dwFlags,
                      &recv_overlapped_.overlapped, NULL);

    if (ret == SOCKET_ERROR) {
        int error = WSAGetLastError();
        if (error != WSA_IO_PENDING) {
            return false;
        }
    }
    return true;
}

bool Connection::IsOnline() { return (endpoint_->IsValid()); }

raptor_error Connection::DoRecvEvent(EventDetail *detail) {
    if (OnRecvEvent(detail->transferred_bytes)) {
        AsyncRecv();
        return RAPTOR_ERROR_NONE;
    }

    return RAPTOR_WINDOWS_ERROR(WSAGetLastError(), "Connection:OnRecvEvent");
}

raptor_error Connection::DoSendEvent(EventDetail *detail) {
    if (OnSendEvent(detail->transferred_bytes)) {
        return RAPTOR_ERROR_NONE;
    }

    return RAPTOR_WINDOWS_ERROR(WSAGetLastError(), "Connection:OnSendEvent");
}

// IOCP Event
bool Connection::OnSendEvent(size_t size) {
    if (size == 0) return false;

    // AutoMutex g(&_snd_mtx);
    send_pending_ = false;
    snd_buffer_.MoveHeader(size);
    if (snd_buffer_.Empty()) {
        return true;
    }
    return AsyncSend();
}

bool Connection::OnRecvEvent(size_t size) {
    (size);

    // AutoMutex g(&_rcv_mtx);
    int recv_bytes = 0;
    int unused_space = 0;
    do {
        char buff[8192] = {0};

        unused_space = sizeof(buff);
        recv_bytes = ::recv((SOCKET)endpoint_->fd_, buff, unused_space, 0);

        if (recv_bytes == 0) return false;
        if (recv_bytes < 0) {
            int err = WSAGetLastError();
            if (WSAEWOULDBLOCK == err) {
                break;
            }
            return false;
        }

        if (proto_) {
            rcv_buffer_.AddSlice(Slice(buff, recv_bytes));
            if (rcv_buffer_.GetBufferLength() >= next_package_size_) {
                if (ParsingProtocol() == -1) {
                    return false;
                }
            }
        } else {
            service_->OnDataReceived(endpoint_, Slice(buff, recv_bytes));
        }

    } while (recv_bytes == unused_space);

    return true;
}

bool Connection::ReadSliceFromRecvBuffer(size_t read_size, Slice &s) {
    size_t cache_size = rcv_buffer_.GetBufferLength();
    if (read_size >= cache_size) {
        s = rcv_buffer_.Merge();
        return true;
    }
    s = rcv_buffer_.GetHeader(read_size);
    return false;
}

int Connection::ParsingProtocol() {
    constexpr size_t header_size = 256;

    size_t cache_size = rcv_buffer_.GetBufferLength();
    int package_counter = 0;

    if (cache_size >= next_package_size_ && next_package_size_ > 0) {
        Slice package = rcv_buffer_.GetHeader(next_package_size_);
        service_->OnDataReceived(endpoint_, package);
        rcv_buffer_.MoveHeader(next_package_size_);
        cache_size -= next_package_size_;
        next_package_size_ = 0;
    }

    while (cache_size > 0) {
        size_t read_size = header_size;
        int pack_len = 0;
        Slice package;
        do {
            bool reach_tail = ReadSliceFromRecvBuffer(read_size, package);
            pack_len = proto_->OnCheckPackageLength(endpoint_, package.begin(), package.size());
            if (pack_len < 0) {
                log_error("Connection: Internal protocol parsing error");
                return -1;
            }

            // equal 0 means we need more data
            if (pack_len == 0) {
                if (reach_tail) {
                    goto done;
                }
                read_size *= 2;
                continue;
            }

            // We got a complete packet with length pack_len
            if (cache_size >= (size_t)pack_len) {
                break;
            }
            next_package_size_ = pack_len;
            goto done;
        } while (true);

        if (package.size() < static_cast<size_t>(pack_len)) {
            package = rcv_buffer_.GetHeader(pack_len);
        } else {
            size_t n = package.size() - pack_len;
            package.RemoveTail(n);
        }
        service_->OnDataReceived(endpoint_, package);
        rcv_buffer_.MoveHeader(pack_len);

        cache_size = rcv_buffer_.GetBufferLength();
        package_counter++;
    }
done:
    return package_counter;
}

} // namespace raptor
